package com.atguigu.state;

/*
*
* 案例需求：通过map算子记录输入的每个数据
* 算子状态 -列表状态
*
* */

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.OperatorStateStore;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

public class Flink02_OperatorListState {
     public static void main(String[] args) {
             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
             env.setParallelism(1);

             //开启检查点，程序失败后会无限重启，连接mysql必须开检查点
             env.enableCheckpointing(2000L);

             //对接数据源
         DataStreamSource<String> ds = env.socketTextStream("hadoop102", 9999);

         //需求：通过map算子记录输入的每个数据
         ds.map(
                 new OperatorListStateMapFunction()
         ).setParallelism(2)
          .addSink(
                 new MyPrintSinkFunction()
         );

         try {
                 env.execute();
             } catch (Exception e) {
                 throw new RuntimeException(e);
             }
         }

    private static class OperatorListStateMapFunction implements MapFunction<String, String>, CheckpointedFunction {

        //定义一个集合，记录map中的所有输入数据
        //1.原始状态
        //List<String> = new ArrayList<>();

        //列表状态
        private ListState<String> listState;//声明一个列表状态用来存中间数据状态
         @Override
        public String map(String value) throws Exception {
            //往状态中添加数据
            listState.add(value);
            //从状态中获取数据
            return listState.get().toString();
        }

        /*
        * 对状态进行持久化存储，基于检查点机制
        *
        * 周期性执行，按照检查点的周期来调用
        * */
        @Override
        public void snapshotState(FunctionSnapshotContext functionSnapshotContext) throws Exception {
            System.out.println("snapshotState........");
        }

        /*
        * 初始化状态，状态的获取 以及 程序重启时状态的还原
        *
        * 程序启动时会调用一次
        *
        * */

        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            System.out.println("initializeState..........");

            //初始化状态
            OperatorStateStore operatorStateStore = context.getOperatorStateStore();
            ListStateDescriptor<String> listStateDesc = new ListStateDescriptor<>("listStateDesc", Types.STRING);
            listState  = operatorStateStore.getListState(listStateDesc);

        }
    }

    private static class MyPrintSinkFunction implements SinkFunction<String> {
        @Override
        public void invoke(String value, Context context) throws Exception {
            if (value.contains("x")){
                throw new RuntimeException("抛出了异常");
            }
            System.out.println("MYPRINT>" + value);
        }
    }
}
